package com.atbeijing.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark02_Req_HotCategoryTop10Session {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
        val sc = new SparkContext(conf)

        val action = sc.textFile("data/user_visit_action.txt")

        // 创建累加器
        val acc = new HotCategoryAccumulator
        // 注册累加器
        sc.register(acc, "HotCategory")

        // 累加器的执行代码一般用在行动算子中，而不是转换算子
        action.foreach(
            line => {
                val datas = line.split("_")
                if ( datas(6) != "-1" ) {
                    // 点击的场合
                    acc.add((datas(6), "click"))
                } else if ( datas(8) != "null" ) {
                    // 下单的场合
                    val ids = datas(8).split(",")
                    ids.foreach(
                        id => {
                            acc.add((id, "order"))
                        }
                    )
                } else if (datas(10) != "null") {
                    // 支付的场合
                    val ids = datas(10).split(",")
                    ids.foreach(
                        id => {
                            acc.add((id, "pay"))
                        }
                    )
                }
            }
        )

        // 获取累加器的值
        val counts: List[HotCategoryCount] = acc.value.map(_._2).toList

        val top10 = counts.sortWith(
            (left, right) => {
                if ( left.clickCnt > right.clickCnt ) {
                    true
                } else if ( left.clickCnt == right.clickCnt ) {
                    if ( left.orderCnt > right.orderCnt ) {
                        true
                    } else if ( left.orderCnt == right.orderCnt ) {
                        left.payCnt > right.payCnt
                    } else {
                        false
                    }
                } else {
                    false
                }
            }
        ).take(10)
        top10.foreach(println)
        // ****************************************************
        // TODO 需求二 : Top10热门品类中每个品类的Top10活跃Session统计
        action.cache()
        val clickDatas = action.filter(
            line => {
                val datas: Array[String] = line.split("_")
                val actionObj = UserVisitAction(
                    datas(0),
                    datas(1).toLong,
                    datas(2),
                    datas(3).toLong,
                    datas(4),
                    datas(5),
                    datas(6).toLong,
                    datas(7).toLong,
                    datas(8),
                    datas(9),
                    datas(10),
                    datas(11),
                    datas(12).toLong
                )

                val ids: List[String] = top10.map(_.cid)
                ids.contains(actionObj.click_category_id.toString) // false
            }
        )

        // 将数据的结构进行转换
        // line => ( (品类,session), 1 )
        // 将转换结构后的数据进行聚合
        //  ( (品类,session), 1 ) =>  ( (品类,session), sum )
        val reduceDatas = clickDatas.map(
            line => {
                val datas = line.split("_")
                ( ( datas(6), datas(2) ), 1 )
            }
        ).reduceByKey(_+_)

        // 将聚合后的结果转换结构
        //  ( (品类,session), sum ) =>  ( 品类, (session, sum) )
        val mapdatas = reduceDatas.map {
            case (( cid, sessionid ), sum) => {
                (cid, (sessionid, sum))
            }
        }

        // 将转换结构后的数据根据品类进行分组
        // (品类， Iter[  (session1, sum1),  (session2, sum2), (session3, sum3)    ])
        val groupDatas: RDD[(String, Iterable[(String, Int)])] = mapdatas.groupByKey()

        // 将分组后的数据根据统计数量进行排序（降序），取前10名
        val top10Session = groupDatas.mapValues(
            iter => {
                iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
            }
        )

        top10Session.collect().foreach(println)


        sc.stop()

    }
    case class UserVisitAction(
          date: String,//用户点击行为的日期
          user_id: Long,//用户的ID
          session_id: String,//Session的ID
          page_id: Long,//某个页面的ID
          action_time: String,//动作的时间点
          search_keyword: String,//用户搜索的关键词
          click_category_id: Long,//某一个商品品类的ID
          click_product_id: Long,//某一个商品的ID
          order_category_ids: String,//一次订单中所有品类的ID集合
          order_product_ids: String,//一次订单中所有商品的ID集合
          pay_category_ids: String,//一次支付中所有品类的ID集合
          pay_product_ids: String,//一次支付中所有商品的ID集合
          city_id: Long//城市 id
      )
    // 样例类中的参数可以被自动当成类的属性，默认参数值不能改，如果想要改，需要明确声明为var
    case class HotCategoryCount(cid:String, var clickCnt:Long, var orderCnt:Long, var payCnt:Long)
    /**
     * 热门品类累加器
     * 1. 继承AccumulatorV2类
     * 2. 定义泛型
     *     IN : 向累加器中输入的数据类型 : (品类，行为类型)
     *     OUT : 累加器的累加结果的类型 : Map[品类，HotCategoryCount]
     *                                Map[品类，(Int, Int, Int))]
     * 3. 重写方法
     */
    class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]] {

        private val hcmap = mutable.Map[String, HotCategoryCount]()

        override def isZero: Boolean = hcmap.isEmpty

        override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]] =  {
            new HotCategoryAccumulator()
        }

        override def reset(): Unit = hcmap.clear()

        override def add(v: (String, String)): Unit = {

            val ( cid, actiontype ) = v

            val hc = hcmap.getOrElse(cid, HotCategoryCount(cid, 0L, 0L, 0L))

            actiontype match {
                case "click" => hc.clickCnt += 1
                case "order" => hc.orderCnt += 1
                case "pay" => hc.payCnt += 1
            }

            hcmap.update(cid, hc)
        }

        override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]]): Unit =  {
            val map = other.value
            map.foreach {
                case ( cid, otherHC ) => {
                    val thishc = hcmap.getOrElse(cid, HotCategoryCount(cid, 0L, 0L, 0L))

                    thishc.clickCnt += otherHC.clickCnt
                    thishc.orderCnt += otherHC.orderCnt
                    thishc.payCnt += otherHC.payCnt

                    hcmap.update(cid, thishc)
                }
            }
        }

        override def value: mutable.Map[String, HotCategoryCount] = hcmap
    }
}
